package top.birdhk.stream;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import top.birdhk.batch.WordCount;



public class StreamWordCount {

    public static void main(String[] args) throws Exception {

        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        // 用parameter tool工具从程序启动参数中提取配置项
//        ParameterTool parameterTool = ParameterTool.fromArgs(args);

//        String host = parameterTool.get("host");
//        int port = parameterTool.getInt("port");


        // 从socket文本流读取数据
        // 使用nc -lk 7777命令
        DataStream<String> inputDataStream = env.socketTextStream("localhost",7777);


        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = inputDataStream.flatMap(new WordCount.MyFlatMapper())
                .keyBy(0)
                .sum(1).setParallelism(3);


        sum.print();

        env.execute();


    }



}
